package cn.gwm.flink.streaming.dwm.charge.function;

import cn.gwm.flink.streaming.config.ChargeConfigEnum;
import cn.gwm.flink.streaming.constant.BaseFields;
import cn.gwm.flink.streaming.constant.VehicleType;
import cn.gwm.flink.streaming.dwm.charge.assist.ChargeFieldDram;
import cn.gwm.flink.streaming.dwm.charge.assist.ChargeSteadyFieldUtil;
import cn.gwm.flink.streaming.dwm.charge.model.ChargeConstant;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.*;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @ClassName ChargePreFunction
 * @Description 充电主题 前置处理
 * 1. 插枪连接的判断 2.非充电字段的过滤 3.经纬度转换处理 4.各字段稳态处理
 * @Author LiangGuang
 * @Date 2023/05/09 19:21
 */
@Slf4j
public class ChargePreFunction extends RichFlatMapFunction<JSONObject, JSONObject> {

    private List<String> fieldList;

    private ChargeConstant constant;
    /**
     * 用于 wavemap 清除字段
     */
    private List<String> keyList = new ArrayList<>();

    public ChargePreFunction(ChargeConstant constant) {
        this.constant = constant;
        fieldList = ChargeSteadyFieldUtil.chargeFieldList(constant.powerType);
    }

    /**
     * 字段稳态保存--> k-字段 v 当前稳态值
     */
    private MapState<String, Integer> steadyMapState;
    /**
     * 缓存数据 缓存中的数据需要进行稳态的判断
     */
    private ListState<JSONObject> listCacheState;

    /**
     * 波动 + 时间  k = 字段-val  v = item_time
     */
    private MapState<String, Long> waveMapState;

    /**
     * 前一条数据时间
     */
    private ValueState<Long> preTimeState;

    /**
     * 是否 在充电主题中 默认false
     */
    private ValueState<Boolean> isChargeTheme;
    /**
     * 最后在主题时间
     */
    private ValueState<Long> lastChargeTheme;

    @Override
    public void open(Configuration parameters) throws Exception {
        steadyMapState = getRuntimeContext().getMapState(new MapStateDescriptor<>("steadyMapState", String.class, Integer.class));
        waveMapState = getRuntimeContext().getMapState(new MapStateDescriptor<>("waveMapState", String.class, Long.class));
        listCacheState = getRuntimeContext().getListState(new ListStateDescriptor<>("listCacheState", JSONObject.class));

        preTimeState = getRuntimeContext().getState(new ValueStateDescriptor<>("preTimeState", Long.class));
        isChargeTheme = getRuntimeContext().getState(new ValueStateDescriptor<>("isChargeTheme", Boolean.class, Boolean.FALSE));
        lastChargeTheme = getRuntimeContext().getState(new ValueStateDescriptor<>("lastChargeTheme", Long.class, System.currentTimeMillis() + 10 * 60 * 1000L));
    }

    @Override
    public void flatMap(JSONObject nowData, Collector<JSONObject> collector) throws Exception {
        // 1.item_time 进行去重--> 不知为何 数据会重复，此处进行校验
        // 2.提前进行判断是否进入主题
        // 3. 减少字段处置
        // 4. 单字段的稳态值处理
        Long preTime = preTimeState.value();
        Long nowTid = nowData.getLong(BaseFields.tid);
        if (preTime == null || nowTid > preTime) {
            preTimeState.update(nowTid);
            // 关键字段为-999999 则过滤
            if (BaseFields.defaultVal.equals(nowData.getStr(constant.getObcConnectStsEv()))
                    && BaseFields.defaultVal.equals(nowData.getStr(constant.getBmsDcchrgConnect()))
                    && BaseFields.defaultVal.equals(nowData.getStr(constant.getObcCclineConnectSts()))) {
                return;
            }
            // 修改 充电相关字段 如果为 默认值则改成0
            connectFieldHandle(nowData);
            // 是否进入充电主题 判定
            lastChargeThemeHandle(preTime, nowData);
            if (isChargeTheme.value() || (nowData.containsKey(BaseFields.REQUEST_TIME) && !BaseFields.defaultVal.equals(nowData.getStr(BaseFields.REQUEST_TIME)))) {
                preHandle(nowData, collector);
            }
        }
    }

    /**
     * 减少字段和 单字段的处理
     */
    public void preHandle(JSONObject nowData, Collector<JSONObject> collector) throws Exception {
        // 减少字段 ，只保留充电相关即可
        nowData = ChargeFieldDram.fieldDram(nowData, constant);
        // 经纬度转换
        // 各字段稳态处理
        fieldSteadyHandle(nowData, collector);
    }


    private void fieldSteadyHandle(JSONObject source, Collector<JSONObject> collector) throws Exception {

        List<JSONObject> list = CollectionUtil.newArrayList(listCacheState.get());
        // 遍历所有需要 进行稳态判断的字段
        for (int i = 0; i < fieldList.size(); i++) {
            fieldSteady(list, source, fieldList.get(i));
        }
        list.add(source);
        // 遍历 缓存
        cacheHandle(list, collector);
    }

    /**
     * 单 字段稳态的判断
     */
    public void fieldSteady(List<JSONObject> list, JSONObject source, String field) throws Exception {
        if (!source.containsKey(field)) {
            return;
        }
        String fieldVal = field + ChargeSteadyFieldUtil.FIELD_STEADY_SUFFIX;
        String fieldFlag = field + ChargeSteadyFieldUtil.FIELD_STEADY_FLAG;
        // 判断是否存在 不存在则默认值
        if (!steadyMapState.contains(field)) {
            steadyMapState.put(field, 0);
        }
        Integer aSteady = steadyMapState.get(field);
        // 对电流单独判断
        Integer nowVal;
        if (field.equals(constant.getBmsBattCurr())) {
            nowVal = source.getDouble(field) >= 0 ? 0 : -1;
        } else {
            nowVal = source.getInt(field);
        }

        if (aSteady.equals(nowVal)) {
            // 标记为true，并且缓存中该字段都为 当前稳态
            list.forEach(item -> {
                if (!item.getBool(fieldFlag)) {
                    item.set(fieldVal, aSteady);
                    item.set(fieldFlag, Boolean.TRUE);
                }
            });
            source.set(fieldVal, aSteady);
            source.set(fieldFlag, Boolean.TRUE);
            removeWaveMapKey(field);
        } else {
            // 即当前不一致了
            String key = field + StrUtil.DASHED + nowVal;
            if (waveMapState.contains(key)) {
                // 已经有波动
                Long firstWaveTime = waveMapState.get(key);
                Long waveTime = Integer.valueOf(ChargeConfigEnum.DATAPERSISTENCE.getValue()) * 1000L;
                if (VehicleType.ES11.equals(source.getStr(VehicleType.VEHICLE_TYPE))) {
                    waveTime = 60 * 1000L;
                }


//                if (ChargeSteadyFieldUtil.connectField(field)) {
//                    // 并且 从1->0 需要进行 3分钟判断
//                    if (Integer.valueOf(steadyMapState.get(field)) != 0 && Integer.valueOf(nowVal) == 0) {
//                        waveTime = Integer.valueOf(ChargeConfigEnum.PLUGINTERVAL.getValue()) * 60 * 1000L;
//                    }
//                }
                if (source.getLong(BaseFields.tid) - firstWaveTime >= waveTime) {
                    // 稳态了，则之前数据全为 当前稳态值，之后的都为现在值
                    list.forEach(item -> {
                        if (!item.getBool(fieldFlag)) {
                            if (item.getLong(BaseFields.tid) < firstWaveTime) {
                                item.set(fieldVal, aSteady);
                            } else {
                                item.set(fieldVal, nowVal);
                            }
                            item.set(fieldFlag, Boolean.TRUE);
                        }
                    });
                    // remove所有field开头的
                    removeWaveMapKey(field);
                    steadyMapState.put(field, nowVal);
                    source.set(fieldVal, nowVal);
                    source.set(fieldFlag, Boolean.TRUE);
                } else {
                    source.set(fieldFlag, Boolean.FALSE);
                }
            } else {
                // 加入 并将其他的改成当前稳态，且清空波动状态
                list.forEach(item -> {
                    if (!item.getBool(fieldFlag)) {
                        item.set(fieldVal, aSteady);
                        item.set(fieldFlag, Boolean.TRUE);
                    }
                });
                removeWaveMapKey(field);
                waveMapState.put(key, source.getLong(BaseFields.tid));
                source.set(fieldFlag, Boolean.FALSE);
            }
        }
    }


    public void cacheHandle(List<JSONObject> list, Collector<JSONObject> collector) throws Exception {
        AtomicInteger ai = new AtomicInteger(0);
        for (int i = 0; i < list.size(); i++) {
            JSONObject item = list.get(i);
            Boolean allFlag = true;
            for (int j = 0; j < fieldList.size(); j++) {
                String field = fieldList.get(j);
                String fieldFlag = field + ChargeSteadyFieldUtil.FIELD_STEADY_FLAG;
                if (!item.getBool(fieldFlag)) {
                    allFlag = false;
                    break;
                }
            }
            if (allFlag) {
                ai.set(i + 1);
                collector.collect(item);
            } else {
                break;
            }
        }
        // 移除ai 之前的缓存
        for (int i = 0; i < ai.get(); i++) {
            list.remove(0);
        }
        listCacheState.clear();
        listCacheState.addAll(list);
    }


    public void removeWaveMapKey(String field) throws Exception {
        keyList.clear();
        waveMapState.keys().forEach(item -> {
            if (item.startsWith(field)) {
                keyList.add(item);
            }
        });
        keyList.forEach(item -> {
            try {
                waveMapState.remove(item);
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        });
    }

    /**
     * @param preTime 上一条数据时间
     * @param source  本条数据
     * @throws Exception
     */
    public void lastChargeThemeHandle(Long preTime, JSONObject source) throws Exception {
        // 如果同时为0，则退出 ，否则更新
        if (source.getInt(constant.getObcConnectStsEv()) == 0 && source.getInt(constant.getBmsDcchrgConnect()) == 0
                && source.getInt(constant.getObcCclineConnectSts()) == 0) {
            // 对比时间，是否 超过 拔枪稳态，多给1分钟

            Long waveTime = (Integer.valueOf(ChargeConfigEnum.PLUGINTERVAL.getValue()) + 1) * 60 * 1000L;
            Long lastTime = lastChargeTheme.value();
            // 但是如果此处是跳变，也需要发送下游-->  todo 处理跳变数据 ？？
            if (preTime != null && (source.getLong(BaseFields.tid) - preTime) > 20 * 1000L) {
                lastChargeTheme.update(source.getLong(BaseFields.tid));
                return;
            }
            if (source.getLong(BaseFields.tid) - lastTime >= waveTime) {
                isChargeTheme.update(Boolean.FALSE);
            }
        } else {
            lastChargeTheme.update(source.getLong(BaseFields.tid));
            if (!isChargeTheme.value()) {
                isChargeTheme.update(Boolean.TRUE);
            }
        }

    }


    public void connectFieldHandle(JSONObject source) {
        if (BaseFields.defaultVal.equals(source.getStr(constant.getObcConnectStsEv()))) {
            source.set(constant.getObcConnectStsEv(), 0);
        }
        if (BaseFields.defaultVal.equals(source.getStr(constant.getBmsDcchrgConnect()))) {
            source.set(constant.getBmsDcchrgConnect(), 0);
        }
        if (BaseFields.defaultVal.equals(source.getStr(constant.getObcCclineConnectSts()))) {
            source.set(constant.getObcCclineConnectSts(), 0);
        }
        if (BaseFields.defaultVal.equals(source.getStr(constant.getBmsBattCurr()))) {
            source.set(constant.getBmsBattCurr(), 0);
        }
        if (BaseFields.defaultVal.equals(source.getStr(constant.getBmsChrgSts()))) {
            source.set(constant.getBmsChrgSts(), 0);
        }
    }


}
